/**注入RocketMQTemplate进行消息发送
 * 发送普通消息：同步（syncSend/convertAndSend）、异步（asyncSend）、单向（sendOneway）、事务（sendMessageInTransaction）、请求响应（sendAndReceive）
 * 发送顺序消息：同步（syncSendOrderly）、异步（asyncSendOrderly）、单向（sendOnewayOrderly）
 * 发送消息参数：基本参数--destination（topic:tag）、message（Message<?>/Object/Collection<T>）、Callback（异步必须）、hashkey（String，Orderly必须）
 *             可选参数--timeout（long）、delaylevel（int）
 */
package com.example.rocketmqspringbootdemo.service.producer.impl;

import com.alibaba.fastjson.JSONObject;
import com.example.rocketmqspringbootdemo.pojo.User;
import com.example.rocketmqspringbootdemo.service.producer.SendService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Slf4j
@Service
public class SendServiceImpl implements SendService {

    @Value("${rocketmq.producer.send-message-timeout}")
    private Long sendMsgTimeout;

    private final RocketMQTemplate rocketMQTemplate;

    @Autowired
    public SendServiceImpl(RocketMQTemplate rocketMQTemplate) {
        this.rocketMQTemplate = rocketMQTemplate;
    }

    /*---------------------------------同步发送------------------------------*/
    //同步发送消息：Message、Collection、Object
    @Override
    public String syncSendStringMessage(String message, String destination) {
        log.info("sendStringMessageSync: message --> {}, destination --> {}", message, destination);
     log.warn("warn message");
     log.error("error meessage");
     log.debug("debug message");
        SendResult result = rocketMQTemplate.syncSend(destination, message);
        /*重载方法：
          syncSend(String destination, Message<?> message, long timeout)
          syncSend(String destination, Message<?> message, long timeout, int delayLevel)
          syncSend(String destination, Collection<T> messages)
          syncSend(String destination, Collection<T> messages, long timeout)
          syncSend(String destination, Object payload)
          syncSend(String destination, Object payload, long timeout)
         */
        return result.getMessageQueue().getTopic() + "----" + result.getSendStatus().toString();
    }

    //同步发送消息：Object
    @Override
    public String syncSendObjectMessage(Object payload, String destination) {
        log.info("sendObjectMessageSync： payload --> {}, destination --> {}",
                JSONObject.toJSONString(payload), destination);
        SendResult result = this.rocketMQTemplate.syncSend(destination, payload);
        return result.getMessageQueue().getTopic() + "----" + result.getSendStatus().toString();
    }

    //同步发送消息：Collection
    @Override
    public String syncSendCollectionMessage(String message, String topic, String tag, Integer batchSize) {
        log.info("sendCollectionMessage：message --> {}, topic --> {}, tag --> {}, batchSize --> {}",
                message, topic, tag, batchSize);

        String destination = getRocketMQDestination(topic, tag);
        Collection<Message<String>> messageList = new ArrayList();
        for (int i = 1; i <= batchSize; i++) {
            String msgStr = message.concat("-") + i;
            Message<String> msg = MessageBuilder.withPayload(msgStr).build();
            messageList.add(msg);
        }
        System.out.println("------------abc");
        SendResult result = this.rocketMQTemplate.syncSend(destination, messageList, sendMsgTimeout);

        return result.getMessageQueue().getTopic() + "----" + result.getSendStatus().toString();
    }

    //同步发送，加TAG---"topic:tagA||tagB"
    @Override
    public String syncSendStringMessageWithTag(String message, String topic, String tag) {
        log.info("sendStringMessageWithTag -- message --> {}, topic --> {}, tag --> {}",
                message, topic, tag);

        String destination = getRocketMQDestination(topic, tag);
        SendResult result = this.rocketMQTemplate.syncSend(destination, message);

        return result.getMessageQueue().getTopic() + "----" + result.getSendStatus().toString();
    }


    @Override
    public String syncSendBatchStringMessageOrderly(String destination) {
        for (int q = 0; q < 4; q++) {
            // send to 4 queues
            List<Message> msgs = new ArrayList<Message>();
            for (int i = 0; i < 10; i++) {
                int msgIndex = q * 10 + i;
                String msg = String.format("Sync Batch Msg#%d to queue: %d", msgIndex, q);

                msgs.add(MessageBuilder.withPayload(msg).
                        setHeader(RocketMQHeaders.KEYS, "KEY_" + msgIndex).build());
            }
            SendResult sr = rocketMQTemplate.syncSendOrderly(destination, msgs, q + "", 60000);
            /*重载方法：
              syncSendOrderly(String destination, Message<?> message, String hashKey)
              syncSendOrderly(String destination, Message<?> message, String hashKey, long timeout)
              syncSendOrderly(String destination, Object payload, String hashKey）
              syncSendOrderly(String destination, Object payload, String hashKey, long timeout)
              syncSendOrderly(String destination, Collection<T> messages, String hashKey)
              asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout,int delayLevel)
              asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout,int delayLevel)

             */
            System.out.println("Send batch messages orderly to queue :" + sr.getMessageQueue().getQueueId());
        }
        return "sync send orderly";
    }

    // Send request in sync mode and receive a reply of String type.
    @Override
    public String syncSendStringReply(String message, String topic, String tag) {
        String destination = getRocketMQDestination(topic, tag);
        log.info("sendStringMessageReply: message --> {}, destination --> {}", message, destination);
        String replyString = rocketMQTemplate.sendAndReceive(destination, message, String.class, 5000);
        System.out.printf("send %s and receive %s %n", message, replyString);
        /*重载方法：
          sendAndReceive(String destination, Message<?> message, Type type)
          sendAndReceive(String destination, Message<?> message, Type type, long timeout)
          sendAndReceive(String destination, Message<?> message, Type type, long timeout, int delayLevel)
          sendAndReceive(String destination, Message<?> message, Type type, String hashKey)
          sendAndReceive(String destination, Message<?> message, Type type, String hashKey, long timeout)
          sendAndReceive(String destination, Message<?> message, Type type, String hashKey,long timeout, int delayLevel)
          sendAndReceive(String destination, Object payload, Type type)
          sendAndReceive(String destination, Object payload, Type type, long timeout)
          sendAndReceive(String destination, Object payload, Type type, long timeout, int delayLevel)
          sendAndReceive(String destination, Object payload, Type type, String hashKey)
          sendAndReceive(String destination, Object payload, Type type, String hashKey, long timeout)
          sendAndReceive(String destination, Object payload, Type type, String hashKey,long timeout, int delayLevel)
         */
        return replyString;
    }

    /*---------------------------------异步发送------------------------------*/
    //异步发送消息
    @Override
    public String asyncSendObjectMessage(Object payload, String destination) {
        log.info("sendObjectMessageAsync -- payload --> {}, destination --> {}",
                JSONObject.toJSONString(payload), destination);
        String[] result = new String[1];
        //发送消息 Start
        this.rocketMQTemplate.asyncSend(destination, payload, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步发送RocketMQ对象消息结果: " + JSONObject.toJSONString(sendResult));
                result[0] = "Success";
            }

            @Override
            public void onException(Throwable e) {
                result[0] = "Fail";
                e.printStackTrace();
            }
        });
        /*重载方法：
          asyncSend(String destination, Message<?> message, SendCallback sendCallback)
          asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout)
          asyncSend(String destination, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel)
          asyncSend(String destination, Object payload, SendCallback sendCallback, long timeout)
          asyncSend(String destination, Collection<T> messages, SendCallback sendCallback)
          asyncSend(String destination, Collection<T> messages, SendCallback sendCallback, long timeout)
         */

//        try {
//            Thread.sleep(2000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

        return result[0];
    }

    //异步发送延迟消息带TAG
    @Override
    public String asyncSendDelayWithTag(Object payload, String topic, String tag, Integer level) {
        log.info("sendDelayWithTag -- payload --> {}, topic --> {}, tag --> {}, level --> {}",
                JSONObject.toJSONString(payload), topic, tag, level);

        String destination = getRocketMQDestination(topic, tag);
        Message<?> message = MessageBuilder.withPayload(payload).build();
        System.out.println("指定Topic指定TAG异步发送延迟对象消息时间: " + LocalDateTime.now());
        String[] result = new String[1];
        this.rocketMQTemplate.asyncSend(destination, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("指定Topic指定TAG异步发送延迟对象消息结果: " + JSONObject.toJSONString(sendResult));
                result[0] = "Success";
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                result[0] = "Fail";
            }
        }, sendMsgTimeout, level);

        try {
            Thread.sleep(12000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return result[0];
    }

    //异步发送顺序消息
    @Override
    public String asyncSendOrderly(String destination) {
        for (int q = 0; q < 4; q++) {
            // send to 4 queues
            List<Message> msgs = new ArrayList<Message>();
            for (int i = 0; i < 10; i++) {
                int msgIndex = q * 10 + i;
                String msg = String.format("Async  Msg#%d to queue: %d", msgIndex, q);
                msgs.add(MessageBuilder.withPayload(msg).build());
            }

            String[] result = new String[1];
            this.rocketMQTemplate.asyncSendOrderly(destination, msgs,q+"", new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("异步顺序消息发送成功!queueID=" + sendResult.getMessageQueue().getQueueId());
                    result[0] = "Success";
                }

                @Override
                public void onException(Throwable e) {
                    e.printStackTrace();
                    result[0] = "Fail";
                }
            }, sendMsgTimeout);
            /*重载方法：
               asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback)
               asyncSendOrderly(String destination, Message<?> message, String hashKey, SendCallback sendCallback,long timeout)
               asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback)
               asyncSendOrderly(String destination, Object payload, String hashKey, SendCallback sendCallback,long timeout)
             */
//            try {
//                Thread.sleep(10000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }

//            System.out.println(result[0]);
        }
        return "Async send orderly......";
    }

    public String asyncSendUserReply(User user, String topic, String tag) {
        String destination = getRocketMQDestination(topic, tag);
        rocketMQTemplate.sendAndReceive(destination, user, new RocketMQLocalRequestCallback<User>() {
            @Override
            public void onSuccess(User message) {
                System.out.printf("Async send user object and receive %s %n", message.toString());
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        }, 5000);
        /*重载方法：
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback)
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout)
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel)
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey)
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout)
          sendAndReceive(String destination, Message<?> message,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, long timeout, int delayLevel)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout)
          sendAndReceive(String destination, Object payload,RocketMQLocalRequestCallback rocketMQLocalRequestCallback, String hashKey, long timeout, int delayLevel)
         */
        return "Async send reply......";
    }

    /*---------------------------------单向发送------------------------------*/
    //同步发送消息：Message、Collection、Object
    @Override
    public void onewaySendString(String message, String destination) {
        log.info("sendStringMessageOneway: message --> {}, destination --> {}", message, destination);
        rocketMQTemplate.sendOneWay(destination, message);
        /*重载方法：
          sendOneWay(String destination, Message<?> message)
          sendOneWay(String destination, Object payload)
         */
    }

    @Override
    public void onewaySendStringOrderly(String destination) {
        for (int q = 0; q < 4; q++) {
            // send to 4 queues
            List<String> msgs = new ArrayList<String>();
            for (int i = 0; i < 10; i++) {
                int msgIndex = q * 10 + i;
                String msg = String.format("Oneway Batch Msg#%d to queue: %d", msgIndex, q);
                msgs.add(msg);
            }
            rocketMQTemplate.sendOneWayOrderly(destination, msgs, q + "");
            /*重载方法：
              sendOneWayOrderly(String destination, Message<?> message, String hashKey)
              sendOneWayOrderly(String destination, Object payload, String hashKey)
             */
        }
    }

    /*---------------------------------事务发送------------------------------*/
    @Override
    public void sendTransactionMessage(String destinnation) {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = MessageBuilder.withPayload("rocketMQTemplate transactional message " + i).
                        setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
                SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
                        destinnation + ":" + tags[i % tags.length], msg, null);
                System.out.printf("[Transaction]send:  msg body = %s , sendResult=%s %n",
                        msg.getPayload(), sendResult.getSendStatus());

                Thread.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /*---------------------------------Util------------------------------*/
    private String getRocketMQDestination(String topic, String tag) {
        return topic.concat(":").concat(tag);
    }
}
